package com.shixj.ef.execute;

import com.shixj.ef.config.FetchConfig;
import com.shixj.ef.execute.base.BaseHttpFetchHandler;
import com.shixj.ef.execute.base.BaseHttpFetchRunner;
import com.shixj.ef.vo.FetchRequest;
import com.shixj.ef.vo.FetchResponse;
import com.shixj.ef.vo.FetchResult;
import org.apache.http.impl.client.CloseableHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by shixj on 2017/9/19.
 */
public class HttpFetchExecutor implements HttpFetchExecute {
    private static final Logger LOGGER = LoggerFactory.getLogger(HttpFetchExecutor.class);
    private CloseableHttpClient httpClient;
    private BlockingQueue<FetchRequest> requestQueue = new LinkedTransferQueue();
    private BlockingQueue<FetchResponse> responseQueue = new LinkedTransferQueue();
    private BlockingQueue<FetchResult> resultQueue = new LinkedTransferQueue();
    private List<FetchResult> fetchResults = new ArrayList<>();
    private FetchConfig config;
    private boolean run=true;
    public HttpFetchExecutor(FetchConfig config) {
        this.config = config;
    }

    public void setHttpClient(CloseableHttpClient httpClient) {
        this.httpClient = httpClient;
    }


    /**
     * 阻塞队列获取返回结果
     *
     * @return
     */
    @Override
    public FetchResult takeResult() {
        try{
            return resultQueue.take();
        }catch (InterruptedException e){
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 添加爬虫结果至阻塞队列
     *
     * @param fetchResult
     */
    @Override
    public void putResult(FetchResult fetchResult) {
        try{
            this.resultQueue.put(fetchResult);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    /**
     * 获取可用请求
     *
     * @return
     */
    @Override
    public FetchRequest takeRequest() {
        try{
            return requestQueue.take();
        }catch (InterruptedException e){
            e.printStackTrace();
            return null;
        }

    }

    /**
     * 向对列添加可用请求
     *
     * @param fetchRequest
     */
    @Override
    public void putRequest(FetchRequest fetchRequest) {
        try{
            this.requestQueue.put(fetchRequest);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    /**
     * 获取可用请求 非阻塞poll
     *
     * @return
     */
    @Override
    public FetchRequest pollRequest() {
        return requestQueue.poll();
    }

    /**
     * 获取可用响应
     *
     * @return
     */
    @Override
    public FetchResponse takeResponse() {
        try{
            return responseQueue.take();
        }catch (InterruptedException e){
            e.printStackTrace();
            return null;
        }
    }
    /**
     * 获取可用响应
     *
     * @return
     */
    @Override
    public FetchResponse pollResponse() {
        return responseQueue.poll();
    }
    /**
     * 获取可用响应
     *
     * @return
     */
    @Override
    public FetchResponse pollResponse(Long timeDelay) {
        try{
            return responseQueue.poll(timeDelay, TimeUnit.MILLISECONDS);
        }catch (InterruptedException e){
            e.printStackTrace();
            return null;
        }
    }
    /**
     * 向对列添加可用响应
     *
     * @param fetchResponse
     */
    @Override
    public void putResponse(FetchResponse fetchResponse) {
        try{
            this.responseQueue.put(fetchResponse);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    @Override
    public boolean isRun() {
        return run;
    }
    @Override
    public void setRun(boolean run) {
        this.run = run;
    }
    @Override
    public FetchConfig getConfig() {
        return config;
    }

    @Override
    public List<FetchResult> getFetchResults() {
        return this.fetchResults;
    }

    /**
     * 添加爬虫结果至list
     *
     * @param fetchResult
     * @return
     */
    @Override
    public void addFetchResult(FetchResult fetchResult) {
        this.fetchResults.add(fetchResult);
    }

    @Override
    public void sendFetch() {
            putRequest(new FetchRequest(this.config.getInitUrl(),this.httpClient));
            for(int i=0;i<this.config.getThreadLimit();i++){
                int index=i+1;
                LOGGER.info("开启爬虫"+index+"：");
                BaseHttpFetchRunner baseHttpFetchRunner = new BaseHttpFetchRunner(this);
                BaseHttpFetchHandler baseHttpFetchHandler = new BaseHttpFetchHandler(this);
                Thread runner = new Thread(baseHttpFetchRunner);
               // LOGGER.info("爬取线程"+index+"：开启");
                runner.start();
                Thread handler = new Thread(baseHttpFetchHandler);
             //   LOGGER.info("分析线程"+index+"：开启");
                handler.start();
            }
    }


}
